%%%====================================================================
%%% File    : pipe_folder.erl
%%% Author  :  <vjache>
%%% Description: This is a simpliest pipe implementation. It may be 
%%%				 considered as a simple reference implementation of 
%%%				 a pipe callback module, and could be used for debug 
%%%				 purposes.
%%%
%%% Created : 24.03.2009
%%%====================================================================

-module(pipe_folder).
 
%%
%% Include files
%%
-include("../include/pipe.hrl").
-include("../include/iter.hrl"). 
%%
%% Exported Functions  
%%
-export([
		 new/1,
		 new/2,
		 open/1,
		 push/2,
		 flush/1,
		 close/1,
		 test/1,
		 test1/1
		 ]).


%%====================================================================
%% API Functions
%%====================================================================

new(NextPipe) ->
	pipe:pipe_from_module(?MODULE, _Param=#folder{}, NextPipe).
new(Folder,NextPipe) when is_record(Folder, folder)->
	pipe:pipe_from_module(?MODULE, Folder, NextPipe).

open({Param,NextPipe}) when is_record(Param, folder) -> %%PipeState
	OpenFun=Param#folder.open_fun,
	AccZero=OpenFun(),
	Countdown=Param#folder.capacity,
    {Param,AccZero,Countdown,pipe:open(NextPipe)}.
push(S,?EOF) ->
	push_acc2next(S);
push({_Param,_Acc,_Countdown=0,_NextPipeState}=S,Item) ->
	push(push_acc2next(S),Item);
push({Param,Acc,Countdown,NextPipeState},Item) ->
    FoldFun=Param#folder.fold_fun,
	{Param,FoldFun(Item,Acc),Countdown-1,NextPipeState}.
flush({_Param,_Acc,_Countdown,_NextPipeState}=S) ->
	push_acc2next(S).
close({Param,Acc,_Countdown,NextPipeState}) ->
	CloseFun=Param#folder.close_fun,
	CloseFun(Acc),
	ok=pipe:close(NextPipeState).

%%====================================================================
%% Test Functions
%%====================================================================

test(L) ->
	S=pipe:open(new(pipe_console:new())),
	{IterState,PipeState}=pipe:push_each(S, iter:from_list(L)),
	iter:close(IterState),
	pipe:close(PipeState).

test1(L) ->
	S=pipe:open(
		new(#folder{
			open_fun=fun()-> dict:new() end,%% Returns a new initial folder
			fold_fun=fun({K,V},AccIn)-> 
							 dict:append(K, V, AccIn);
					 	(?EOF,AccIn)->
							AccIn
					 end,%% Folds Item with AccIn and retuns new one
			close_fun=fun(_Acc)-> ok end,%% Releases resources of folder if any
			capacity=10 %% Returns folder capacity (-1 denotes Infinity). If exhausted then accumulated value passe to next pipe and folder destroyed.
		},pipe_console:new())),
	{IterState,PipeState}=pipe:push_each(S, iter:from_list(L)),
	iter:close(IterState),
	pipe:close(PipeState).


%%====================================================================
%% Local Functions
%%====================================================================
push_acc2next({Param,?EOF,Countdown,NextPipeState}=_S) ->
	{Param,?EOF,Countdown,pipe:push(NextPipeState,?EOF)};
push_acc2next({Param,Acc,_,NextPipeState}) ->
	ResultFun=Param#folder.result_fun,
	NextPipeNewState=case ResultFun(Acc) of	
						 ResList when is_list(ResList) ->
							 lists:foldl(fun(Item,NextPipeStateIn)->
										 pipe:push(NextPipeStateIn,Item)
								 end, NextPipeState, ResList);
						 ResIter when is_record(ResIter, iter) ->
							 {FinalIterState,FinalPipeState}=pipe:push_each(NextPipeState, ResIter),
							 iter:close(FinalIterState),
							 FinalPipeState
					 end,
	CloseFun=Param#folder.close_fun,
	CloseFun(Acc),
	OpenFun=Param#folder.open_fun,
	AccZero=OpenFun(),
	Countdown=Param#folder.capacity,
	{Param,AccZero,Countdown,NextPipeNewState}.

